package com.iot.app.flink.util;

import com.iot.app.flink.entity.POITrafficData;
import com.iot.app.flink.entity.TotalTrafficData;
import com.iot.app.flink.entity.WindowTrafficData;
import com.iot.app.flink.vo.AggregateKey;
import com.iot.app.flink.vo.IoTData;
import com.iot.app.flink.vo.POIData;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;

import javax.annotation.Nullable;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.TimeZone;


public class IotDataStreamUtils {

    private static final FastDateFormat formatter = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss",
            TimeZone.getTimeZone("MST"));

    public static FlatMapFunction<String, IoTData> streamToIotDataMap() {
        return new FlatMapFunction<String, IoTData>() {

            private static final long serialVersionUID = 1L;
            private transient ObjectMapper jsonParser;

            @Override
            public void flatMap(String in, Collector<IoTData> out) throws Exception {

                if (jsonParser == null) {
                    jsonParser = new ObjectMapper();
                }

                JsonNode jsonNode = jsonParser.readValue(in, JsonNode.class);
                boolean isIotData = jsonNode.has("vehicleId") &&
                        jsonNode.has("vehicleType") && jsonNode.has("routeId") &&
                        jsonNode.has("latitude") && jsonNode.has("longitude") &&
                        jsonNode.has("timestamp") && jsonNode.has("speed") &&
                        jsonNode.has("fuelLevel");

                boolean isZeroLength = jsonNode.get("vehicleId").asText().isEmpty() || jsonNode.get("routeId").asText().isEmpty() ||
                        jsonNode.get("latitude").asText().isEmpty() || jsonNode.get("longitude").asText().isEmpty() ||
                        jsonNode.get("timestamp").asText().isEmpty() || jsonNode.get("speed").asText().isEmpty() ||
                        jsonNode.get("fuelLevel").asText().isEmpty();

                if (isZeroLength) {
                    System.out.println("iotdata null length");
                }

                if (isIotData) {
                    out.collect(new IoTData(jsonNode.get("vehicleId")
                            .asText(), jsonNode.get("vehicleType")
                            .asText(), jsonNode.get("routeId")
                            .asText(), jsonNode.get("latitude")
                            .asText(), jsonNode.get("longitude")
                            .asText(),
                            formatter.parse(jsonNode.get("timestamp")
                                    .asText()), jsonNode.get("speed")
                            .asDouble(), jsonNode.get("fuelLevel")
                            .asDouble()));

                }
            }
        };
    }

    public static FlatMapFunction<Tuple2<AggregateKey, Long>, TotalTrafficData> streamToTotalTrafficDataMap( ) {
        return new FlatMapFunction<Tuple2<AggregateKey, Long>, TotalTrafficData>() {
            @Override
            public void flatMap(Tuple2<AggregateKey, Long> in, Collector<TotalTrafficData> out) throws Exception {
                TotalTrafficData ele = new TotalTrafficData();
                ele.setRouteId(in.f0.getRouteId());
                ele.setVehicleType(in.f0.getVehicleType());
                ele.setTimeStamp(new Date());
                ele.setRecordDate(new SimpleDateFormat("yyyy-MM-dd").format(new Date()));
                ele.setTotalCount(in.f1);
                out.collect(ele);
            }
        };
    }

    public static FlatMapFunction<Tuple2<AggregateKey, Long>, WindowTrafficData> streamToWindowTrafficDataMap() {
        return new FlatMapFunction<Tuple2<AggregateKey, Long>, WindowTrafficData>() {
            @Override
            public void flatMap(Tuple2<AggregateKey, Long> in, Collector<WindowTrafficData> out) throws Exception {
                WindowTrafficData ele = new WindowTrafficData();
                ele.setRouteId(in.f0.getRouteId());
                ele.setVehicleType(in.f0.getVehicleType());
                ele.setTimeStamp(new Date());
                ele.setRecordDate(new SimpleDateFormat("yyyy-MM-dd").format(new Date()));
                ele.setTotalCount(in.f1);
                out.collect(ele);
            }
        };
    }

    public static FlatMapFunction<Tuple2<IoTData, POIData>, POITrafficData> streamToPOITrafficDataMap() {
        return new FlatMapFunction<Tuple2<IoTData, POIData>, POITrafficData>() {
            @Override
            public void flatMap(Tuple2<IoTData, POIData> in, Collector<POITrafficData> out) throws Exception {
                POITrafficData ele = new POITrafficData();
                ele.setVehicleId(in.f0.getVehicleId());
                ele.setVehicleType(in.f0.getVehicleType());
                ele.setTimeStamp(in.f0.getTimestamp());
                double distance = GeoDistanceCalculator.getDistance(Double.valueOf(in.f0.getLatitude()).doubleValue(),
                        Double.valueOf(in.f0.getLongitude()).doubleValue(), in.f1.getLatitude(), in.f1.getLongitude());
                ele.setDistance(distance);
                out.collect(ele);
            }
        };
    }

    public static KeySelector<IoTData, String> iotDatakeySelector() {
        return new KeySelector<IoTData, String>() {
            @Override
            public String getKey(IoTData ioTData) throws Exception {
                return ioTData.getVehicleId();
            }
        };
    }

    public static ReduceFunction<IoTData> dedupByKey() {
        return new ReduceFunction<IoTData>() {
            @Override
            public IoTData reduce(IoTData ioTData, IoTData t1) throws Exception {
                return ioTData;
            }
        };
    }

    public static KeySelector<Tuple2<AggregateKey, Long>, AggregateKey> aggregateKeySelector() {
        return new KeySelector<Tuple2<AggregateKey, Long>, AggregateKey>() {
            @Override
            public AggregateKey getKey(Tuple2<AggregateKey, Long> value) throws Exception {
                return value.f0;
            }
        };
    }

    public static KeyedStream<Tuple2<AggregateKey, Long>, AggregateKey> tokeyedStreamForCounting(DataStream<IoTData> dataStream) {
        return dataStream.flatMap(new FlatMapFunction<IoTData, Tuple2<AggregateKey, Long>>() {
            @Override
            public void flatMap(IoTData ioTData, Collector<Tuple2<AggregateKey, Long>> out) throws Exception {
                out.collect(Tuple2.of(new AggregateKey(ioTData.getRouteId(), ioTData.getVehicleType()), new Long(1)));
            }
        }).keyBy(IotDataStreamUtils.aggregateKeySelector());
    }

    public static AssignerWithPeriodicWatermarks<IoTData> eventTimeExtractor() {
        return new AssignerWithPeriodicWatermarks<IoTData>() {

            private long currentMaxEventTime = 0L;
            private long timeout = 1000; // 最大允许的乱序时间 10 秒

            FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");


            @Override
            public long extractTimestamp(IoTData ioTData, long previousElementTimestamp) {
               long currentEventTime = ioTData.getTimestamp().getTime();
                currentMaxEventTime = Math.max(currentEventTime, currentMaxEventTime);
//                System.out.println("event =" + ioTData.getVehicleId()
//                                    + "|" + dateFormat.format(ioTData.getTimestamp().getTime())
//                                    + "|" + dateFormat.format(currentMaxEventTime)
//                                    + "|" + dateFormat.format(getCurrentWatermark().getTimestamp())
//                                    );

                return currentEventTime;
            }

            @Nullable
            @Override
            public Watermark getCurrentWatermark() {

                return new Watermark(currentMaxEventTime - timeout);
            }
        };
    }

    public static CoFlatMapFunction<IoTData, Map<String, String>, Tuple2<IoTData, POIData>> streamToFilterByPoi() {
        return new CoFlatMapFunction<IoTData, Map<String, String>, Tuple2<IoTData, POIData>>() {

            private POIData poiData = null;
            private Logger logger = LoggerFactory.getLogger(this.getClass());

            @Override
            public void flatMap1(IoTData in, Collector<Tuple2<IoTData, POIData>> out) throws Exception {

                if (poiData != null) {
                    if (in.getRouteId().equals(poiData.getRouteId()) &&
                            in.getVehicleType().equals(poiData.getVehicleType()))
                    {
                        Double latitude;
                        Double longitude;

                        try {
                            latitude = Double.valueOf(in.getLatitude());
                        } catch ( NumberFormatException e) {
                            logger.error("in connect latitude on iotdata" + e);
                            return;
                        }

                        try {
                            longitude = Double.valueOf(in.getLongitude());
                        } catch ( NumberFormatException e) {
                            logger.error("in connect longitude on iotdata" + e);
                            return;
                        }

                        if (GeoDistanceCalculator.isInPOIRadius(latitude, longitude, poiData.getLatitude(),
                                poiData.getLongitude(), poiData.getRadius())) {

                            IoTData ioTData = new IoTData(in.getVehicleId(),
                                    in.getVehicleType(),
                                    in.getRouteId(),
                                    in.getLatitude(),
                                    in.getLongitude(),
                                    in.getTimestamp(),
                            in.getSpeed(),
                            in.getFuelLevel());

                            POIData poiDataOut = new POIData(poiData.getLatitude(),
                                    poiData.getLongitude(),
                                    poiData.getRadius(),
                                    poiData.getRouteId(),
                                    poiData.getVehicleType());

                            out.collect(Tuple2.of(ioTData, poiDataOut));
                        }
                    }
                }
            }

            @Override
            public void flatMap2(Map<String, String> in, Collector<Tuple2<IoTData, POIData>> out) throws Exception {

                if (poiData == null) {
                    poiData =  new POIData();
                }

                if (in.containsKey("latitude")) {
                    String value = in.get("latitude");
                    try {
                        poiData.setLatitude(Double.valueOf(value));
                    } catch (NumberFormatException e) {
                        logger.error("latitude" + value +  e );
                    }
                }

                if (in.containsKey("longitude")) {
                    String value = in.get("longitude");
                    try {
                        poiData.setLongitude(Double.valueOf(value));
                    } catch (NumberFormatException e) {
                        logger.error("longitude" + value +  e );
                    }
                }

                if (in.containsKey("radius")) {
                    String value = in.get("radius");
                    try {
                        poiData.setRadius(Double.valueOf(value));
                    } catch (NumberFormatException e) {
                        logger.error("radius" + value +  e );
                    }
                }

                if (in.containsKey("routeId")) {
                    String value = in.get("routeId");
                    poiData.setRouteId(value);
                }

                if (in.containsKey("vehicleType")) {
                    String value = in.get("vehicleType");
                    poiData.setVehicleType(value);
                }
            }
        };

    }
}
